package thss.mro.bolt;

import java.util.Map;

import thss.mro.manager.ProtocolManager;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

/*
 * 包的识别根据端口号识别
 * 使用thss.mro.util.ProtocolType 资源类识别
* */
public class PacketIdentify extends BaseRichBolt {

	OutputCollector _collector;
	
	@Override
	public void execute(Tuple tuple) {
		int port = (Integer)tuple.get("port");
		String protocolType = ProtocolManager.getProtocolType(port);
		String headId = ProtocolManager.getHeadTplId(protocolType);
		byte[] data =  (byte[])tuple.get("data");
		_collector.emit(tuple, new Values(headId,data));//向下游发送协议类型与数据包
		_collector.ack(tuple);
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		_collector = collector;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("headId","data"));
	}

}
